package com.sdemo.utils.mq.kafka;

import java.util.Properties;

import org.apache.log4j.Logger;

import com.sdemo.utils.mq.MsgProducer;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
 
public class KafkaMsgProducer   implements MsgProducer {
	private static final Logger log=Logger.getLogger(KafkaMsgProducer.class);
	private static final String brokerList= null;//ConfigUtil.getString("kafka.broker.list", "172.16.0.105:9092");
    private  Producer<String, byte[]> producer;
	
    public KafkaMsgProducer() {
    	
		Properties props = new Properties();
	    props.put("key.serializer.class", "kafka.serializer.StringEncoder");
	    props.put("metadata.broker.list", brokerList);  
	    props.put("producer.type", "async");
	    props.put("queue.enqueue.timeout.ms", "-1");
	    props.put("batch.num.messages", "2000");
	    props.put("compression.codec", "1");
	    props.put("request.required.acks", "0");
	    props.put("request.timeout.ms", "3000");
	    producer = new kafka.javaapi.producer.Producer<String,byte[]>(new ProducerConfig(props));  
	}
    
    /**
     * 发送消息
     */
	@Override
	public boolean send(String topic,String key,byte[] value) {
		KeyedMessage<String, byte[]> msgRecord=new KeyedMessage<String, byte[]>(topic, key, value);
	    producer.send(msgRecord);
		return true;
	}

 
	public void close(){
		if(null !=producer){
			log.info("close");
			producer.close();
		}
	}

}
